package day02.operator.transformation

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Author wsl
 * @Description
 */
object Value_Map {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc = new SparkContext(conf)

    //map 1个参数
    sc.textFile("sparkcore/input/1.txt")
      .map(
        line => {
          val word: Array[String] = line.split(" ")
          word.mkString(",")
          //word(1)
        }
      )
      .collect().foreach(println)

    val Rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5), 2)

    Rdd.map(_ * 2).collect().foreach(println)

    //mapPartitions  1个参数
    Rdd.mapPartitions(iters => iters.map(t => t * 2))

    Rdd.mapPartitions(_.map(_ * 2)).collect().foreach(println)

    //每个分区Max
    Rdd
      .mapPartitions(iters => List(iters.max).iterator)
      .collect().foreach(println)

    //mapPartitionsWithIndex  2个参数
    Rdd
      .mapPartitionsWithIndex(
        (index,iters)=>{
          iters.map((index,_))
        }
      )

    Rdd
      .mapPartitionsWithIndex((index, iters) => iters.map((index, _)))
      .collect().foreach(println)

    Rdd.mapPartitionsWithIndex(
      (index, iters) => {
        if (index == 2) {
          iters
        } else {
          Nil.iterator
        }
      }
    ).collect().foreach(println)


    //flatmap 1进 迭代器出
    sc.makeRDD(List(List(1, 2), List(3, 4), List(5, 6), List(7)), 2)
      .flatMap(list => list)
      .collect().foreach(println)

    sc.makeRDD(List("hello spark", "hello flink"))
      .flatMap(_.split(" "))
      .collect().foreach(println)

    sc.textFile("sparkcore/input/agent.txt")
      .flatMap(_.split(" "))
      .collect().foreach(println)


    //glom 将RDD中每一个分区变成一个数组
    Rdd
      .glom()
      .collect().foreach(elem => println(elem.mkString(","))) //打印每个分区

    //每个分区Max
    Rdd
      .glom()
      .map(_.max)
      .collect().foreach(println)

    //filter
    Rdd
      .filter(_ % 2 == 0)
      .collect().foreach(println)

    sc.makeRDD(List("hello", "spark", "hi"))
      .filter(_.charAt(0) == 'h')
      .collect().foreach(println)

    sc.stop()
  }
}
